#include "ThreadPoolOfC.h"
#include <pthread.h>
#include <stdlib.h>
#include <stdio.h>
#include <string.h>
#include <unistd.h>

const int NUMBER = 2;

// 任务结构体
// 在 C 语言中结构体的定义是 struct 结构体名字。结构体的使用是 `struct 结构体名字 变量名`。Cpp 中则不需要 `struct` 关键字了所以是 `结构体名字 变量名`。
// 所以在 C 语言中使用结构体为了简化编码，我们一般会像下面一样使用关键字 `typedef` 来给结构体取别名。
// 如下面这样在定义申明时就直接取别名一样，Task 可以代替 struct Task。虽然还是叫 Task 但没有影响。 我们也可以不在定义时声明而是在后面使用 `typedef struct Task Task`来申明是一样的
typedef struct Task
{
    void (*function)(void *arg);
    void *arg;
} Task;

// 线程池结构体
struct ThreadPool
{
    // 任务队列
    Task *taskeQueue;
    // 任务队列的容量
    int queueCapacity;
    // 任务队列的大小
    int queueSize;
    // 任务队列对头 -> 取数据
    int queueFront;
    // 任务队列队尾 -> 放数据
    int queueRear;

    // 线程池
    // 管理者线程 ID
    pthread_t managerId;
    // 工作者线程 ID 数组
    pthread_t *workIds;

    // 线程池容纳范围：最小线程数
    int minNum;
    // 线程池容纳范围：最大线程数
    int maxNum;
    // 工作中的线程数：繁忙的线程数
    int busyNum;
    // 存活的线程数：包括繁忙的和闲置的
    int liveNum;
    // 需要退出或杀死结束的线程数
    int exitNum;
    // 线程池锁：是针对整个线程池操作时需要保护的资源锁，这里实际上线程池中操作的数据主要时任务队列 taskeQueue，那么和任务队列有关的都应该使用这个锁。
    pthread_mutex_t mutexPool;
    // 工作线程锁：当前线程中，线程工作或停止都需要针对 busyNum 进行修改，会比较频繁，所以如果依旧使用线程池锁来处理，会导致其他操作倍不必要的锁住导致性能问题。所以单独对他进行一个锁。
    pthread_mutex_t mutexBusy;
    // 任务队列是否满了
    pthread_cond_t notEmpty;
    // 任务队列是否空了
    pthread_cond_t notFull;
    // 工作线程是否完全退出了
    pthread_cond_t notLive;
    // 线程池状态：是否需要销毁线程池，0 不需要 1 销毁
    int shutdown;
};

ThreadPool *threadPoolCreate(int min, int max, int queueSize)
{
    // 创建线程池对象
    ThreadPool *threadPool = (ThreadPool *)malloc(sizeof(ThreadPool));
    if (threadPool == NULL)
    {
        printf("创建线程池失败！\n");
        return NULL;
    }

    // 初始化线程池对象
    do
    {
        // 初始化任务队相关变量
        threadPool->queueCapacity = queueSize;
        threadPool->queueSize = 0;
        threadPool->queueFront = 0;
        threadPool->queueRear = 0;
        // 初始化线程池相关变量
        threadPool->minNum = min;
        threadPool->maxNum = max;
        threadPool->liveNum = min;
        threadPool->busyNum = 0;
        threadPool->exitNum = 0;
        threadPool->shutdown = 0;

        // 创建任务队列
        threadPool->taskeQueue = (Task *)malloc(sizeof(Task) * queueSize);
        if (threadPool->taskeQueue == NULL)
        {
            printf("创建任务队列失败！\n");
            break;
        }

        // 创建工作线程 ID 存放数组，并初始化数组内存，按最大可存放数量创建，但初始化时只按最小线程数创建存活线程
        int workIdSize = sizeof(pthread_t) * max;
        threadPool->workIds = (pthread_t *)malloc(workIdSize);
        if (threadPool->workIds == NULL)
        {
            printf("创建工作线程 ID 存放数组失败！\n");
            break;
        }
        memset(threadPool->workIds, 0, workIdSize);

        // 初始化同步锁
        if (pthread_mutex_init(&threadPool->mutexPool, NULL) != 0 || pthread_mutex_init(&threadPool->mutexBusy, NULL) != 0)
        {
            printf("初始化同步锁失败！\n");
            break;
        }

        // 初始化条件变量
        if (pthread_cond_init(&threadPool->notEmpty, NULL) != 0 ||
            pthread_cond_init(&threadPool->notFull, NULL) != 0 ||
            pthread_cond_init(&threadPool->notLive, NULL) != 0)
        {
            printf("初始化条件变量失败！\n");
            break;
        }

        // 创建管理处理线程
        int managerThreadRet = pthread_create(&threadPool->managerId, NULL, managerProcess, threadPool);
        if (managerThreadRet != 0)
        {
            printf("创建管理者处理线程失败\n");
            break;
        }

        // 创建最小工作处理线程
        int workCount = 0;
        for (int i = 0; i < min; ++i)
        {
            int workThreadRet = pthread_create(&threadPool->workIds[i], NULL, workerProcess, threadPool);
            if (workThreadRet != 0)
                break;
            pthread_detach(threadPool->workIds[i]);
            workCount++;
        }
        if (workCount < min)
        {
            printf("创建最小工作处理线程失败\n");
            break;
        }

        return threadPool;
    } while (false);

    // 如果在上面没有直接返回，说明某个环节初始化失败了，线程池创建有问题，释放所有已创建的资源
    if (threadPool->workIds != NULL)
    {
        free(threadPool->workIds);
    }
    if (threadPool->taskeQueue != NULL)
    {
        free(threadPool->taskeQueue);
    }
    if (threadPool != NULL)
    {
        free(threadPool);
    }

    return NULL;
}

int threadPoolDestroy(ThreadPool* threadPool)
{
    if(threadPool == NULL)
        return -1;
    // 关闭线程
    threadPool->shutdown = 1;

    // 回收管理者线程
    pthread_join(threadPool->managerId, NULL);

    // 唤醒工作线程
    for(int i = 0; i < threadPool->liveNum; ++i)
    {
        pthread_cond_signal(&threadPool->notEmpty);
    }
    while(threadPool->liveNum) {
        pthread_cond_wait(&threadPool->notLive, &threadPool->mutexPool);
    }
    // for(int i = 0; i < threadPool->maxNum; ++i)
    // {
    //     pthread_t workeId = threadPool->workIds[i];
    //     if(workeId != 0)
    //     {
    //         pthread_join(workeId, NULL);
    //     }
    // }

    if(threadPool->taskeQueue != NULL)
    {
        free(threadPool->taskeQueue);
        threadPool->taskeQueue = NULL;
    }

    if(threadPool->workIds != NULL)
    {
        free(threadPool->workIds);
        threadPool->workIds = NULL;
    }

    pthread_mutex_destroy(&threadPool->mutexBusy);
    pthread_mutex_destroy(&threadPool->mutexPool);
    pthread_cond_destroy(&threadPool->notEmpty);
    pthread_cond_destroy(&threadPool->notFull);
    pthread_cond_destroy(&threadPool->notLive);

    free(threadPool);
    threadPool = NULL;
    
    return 0;
}

void threadPoolAdd(ThreadPool * threadPool, void (*function)(void *), void *arg)
{
    pthread_mutex_lock(&threadPool->mutexPool);
    while(threadPool->queueSize >= threadPool->queueCapacity && !threadPool->shutdown)
    {
        pthread_cond_wait(&threadPool->notFull, &threadPool->mutexPool);
    }
    if(!threadPool->shutdown)
    {
        threadPool->taskeQueue[threadPool->queueRear].function = function;
        threadPool->taskeQueue[threadPool->queueRear].arg = arg;
        threadPool->queueRear = (threadPool->queueRear + 1) % threadPool->queueCapacity;
        threadPool->queueSize++;
        pthread_cond_signal(&threadPool->notEmpty);
    }
    pthread_mutex_unlock(&threadPool->mutexPool);
}

int threadPoolBusyNum(ThreadPool* threadPool)
{
    pthread_mutex_lock(&threadPool->mutexBusy);
    int busyNum = threadPool->busyNum;
    pthread_mutex_unlock(&threadPool->mutexBusy);
    return busyNum;
}

int threadPoolLiveNum(ThreadPool* threadPool)
{
    pthread_mutex_lock(&threadPool->mutexPool);
    int liveNum = threadPool->liveNum;
    pthread_mutex_unlock(&threadPool->mutexPool);
    return liveNum;
}

void *managerProcess(void *arg)
{
    ThreadPool *threadPool = (ThreadPool *)arg;
    while (!threadPool->shutdown)
    {
        sleep(3);
        printf("管理线程: ID=%ld，开始工作\n", pthread_self());

        pthread_mutex_lock(&threadPool->mutexPool);
        int queueSize = threadPool->queueSize;
        int liveNum = threadPool->liveNum;
        pthread_mutex_unlock(&threadPool->mutexPool);

        pthread_mutex_lock(&threadPool->mutexBusy);
        int busyNum = threadPool->busyNum;
        pthread_mutex_unlock(&threadPool->mutexBusy);

        // 添加线程
        // 添加线程的策略，目前示例简单定义，但可以使用更优策略：
        //  * 比如开辟线程的个数不时固定的 NUMBER 而是更具任务两来预测，
        //  * 比如合适增加线程可以不必每次都在线程不够时久立马添加 liveNum < queueSize，比如存货线程有10个，任务有11个，这个久会增加这显然会频繁创建。可以选择任务超过存货百分之50在增加
        if(liveNum < queueSize && liveNum < threadPool->maxNum) 
        {   
            pthread_mutex_lock(&threadPool->mutexPool);
            int counter = 0;
            for(int i = 0; i < threadPool->maxNum; ++i)
            {
                if(counter >= NUMBER)
                    break;
                if(threadPool->liveNum >= threadPool->maxNum)
                    break;
                if(threadPool->workIds[i] != 0)
                    break;

                int workThreadRet = pthread_create(&threadPool->workIds[i], NULL, workerProcess, threadPool);
                if (workThreadRet != 0)
                    break;
                pthread_detach(threadPool->workIds[i]);
                counter++;
                threadPool->liveNum++;
            }
            pthread_mutex_unlock(&threadPool->mutexPool);
        }

        // 销毁线程
        // 销毁线程的策略，目前示例简单定义，但可以使用更优策略：
        // 比如使用超时管理
        if(busyNum * 2 < liveNum && liveNum > threadPool->minNum)
        {
            pthread_mutex_lock(&threadPool->mutexPool);
            threadPool->exitNum = NUMBER;
            pthread_mutex_unlock(&threadPool->mutexPool);
            for(int i = 0; i < NUMBER; ++i)
            {
                pthread_cond_signal(&threadPool->notEmpty);
            }
        }
    }

    return NULL;
}

void *workerProcess(void *arg)
{
    ThreadPool *threadPool = (ThreadPool *)arg;
    // while (!threadPool->shutdown)
    while (true)
    {
        pthread_mutex_lock(&threadPool->mutexPool);
        // 判断当前任务队列是否有任务需要执行，如果没有则挂起等待
        while (threadPool->queueSize == 0 && !threadPool->shutdown)
        {
            // 挂起等待有任务提交再重新唤醒执行
            pthread_cond_wait(&threadPool->notEmpty, &threadPool->mutexPool);
            if(threadPool->exitNum > 0)
            {
                threadPool->exitNum--;
                if(threadPool->liveNum > threadPool->minNum) {
                    printf("工作线程: ID=%ld，存活线程过多，销毁工作线程\n", pthread_self());
                    exitWorkerFromThreadPool(threadPool);
                    threadPool->liveNum--;
                    pthread_mutex_unlock(&threadPool->mutexPool);
                    pthread_exit(NULL);
                }
            }
        }
        // 如果任务池有任务或者等待时有任务被提交，那么我们就会进入到这里开始执行任务，在执行之前先判断线程池是否被关闭。
        if (threadPool->shutdown)
        {
            printf("工作线程: ID=%ld，线程池关闭，销毁工作线程\n", pthread_self());
            exitWorkerFromThreadPool(threadPool);
            threadPool->liveNum--;
            pthread_cond_broadcast(&threadPool->notLive);
            // 如果线程池关闭了，那么就需要退出当前线程，在那之前需要释放当前获得的锁。
            pthread_mutex_unlock(&threadPool->mutexPool);
            pthread_exit(NULL);
        }
        // 线程池没有关闭，那么去从任务队列里取任务
        // Task task = threadPool->taskeQueue[threadPool->queueFront];
        Task task;
        task.function = threadPool->taskeQueue[threadPool->queueFront].function;
        task.arg = threadPool->taskeQueue[threadPool->queueFront].arg;
        threadPool->queueFront = (threadPool->queueFront + 1) % threadPool->queueCapacity;
        threadPool->queueSize--;

        // 唤醒生产者，可以添加任务了
        pthread_cond_signal(&threadPool->notFull);

        pthread_mutex_unlock(&threadPool->mutexPool);

        // 繁忙线程数加一
        pthread_mutex_lock(&threadPool->mutexBusy);
        threadPool->busyNum++;
        pthread_mutex_unlock(&threadPool->mutexBusy);

        printf("工作线程: ID=%ld，开始工作\n", pthread_self());

        // 执行任务
        task.function(task.arg);
        free(task.arg);
        task.arg = NULL;

        printf("工作线程: ID=%ld，结束工作\n", pthread_self());

        // 繁忙线程数减一
        pthread_mutex_lock(&threadPool->mutexBusy);
        threadPool->busyNum--;
        pthread_mutex_unlock(&threadPool->mutexBusy);
    }
    return NULL;
}

void exitWorkerFromThreadPool(ThreadPool *threadPool)
{
    pthread_t tid = pthread_self();
    for(int i = 0; i < threadPool->maxNum; ++i)
    {
        if(tid == threadPool->workIds[i])
        {
            threadPool->workIds[i] = 0;
            printf("工作线程: ID=%ld，退出销毁\n", tid);
            break;
        }
    }
}

void testAction(void* agr);

void threadPoolDemoOfC()
{
    printf("线程池使用案例（基于 C 标准库）开始\n");
    ThreadPool *threadPool = threadPoolCreate(1, 10, 100);

    for(int i = 0; i < 100; ++i)
    {
        int* value = (int*) malloc(sizeof(int));
        *value = i;
        threadPoolAdd(threadPool, testAction, value);
    }

    sleep(10);

    threadPoolDestroy(threadPool);
    printf("线程池使用案例（基于 C 标准库）结束\n");
}

void testAction(void* agr)
{
    int value = *((int*) agr);
    printf("测试任务: ID=%ld，当前数据：%d\n", pthread_self(), value);
    sleep(1);
}